package com.examples.redis.binlog;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.descriptors.*;

public class FlinkRedisCDCExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 配置Debezium连接器以读取Redis的CDC数据
        tableEnv.executeSql(
                "CREATE TABLE redis_binlog (" +
                        "   `before` ROW<>," +
                        "   `after` ROW<>," +
                        "   `source` ROW<>" +
                        ") WITH (" +
                        "   'connector' = 'debezium'," +
                        "   'format' = 'json'," +
                        "   'debezium.snapshot.mode' = 'initial'," +
                        "   'debezium.io.kafka.bootstrap.servers' = 'kafka-server:9092'," +
                        "   'debezium.io.kafka.topic' = 'redis-binlog-topic'," +
                        "   'scan.startup.mode' = 'earliest-offset'" +
                        ")"
        );

        // 在这里处理数据
        tableEnv.executeSql("INSERT INTO ... SELECT * FROM redis_binlog");

        env.execute("Flink Redis CDC Example");
    }
}
